package cn.ycc1.boot3redis.service;

import cn.ycc1.boot3redis.entity.Message;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.RedisConnectionFailureException;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 消息消费者服务
 * @author ycc
 * @date 2025/11/20
 */
@Service
public class MessageConsumerService {

    private final RedisTemplate<String, Object> redisTemplate;
    private final String messageQueueName;
    private volatile boolean running = false;
    private Thread consumerThread;

    private final AtomicLong processedCount = new AtomicLong(0);
    private final AtomicLong errorCount = new AtomicLong(0);
    private volatile long lastProcessTime = 0;

    public MessageConsumerService(RedisTemplate<String, Object> redisTemplate,
                                  @Value("${app.redis.queue.name}") String messageQueueName) {
        this.redisTemplate = redisTemplate;
        this.messageQueueName = messageQueueName;
    }

    /**
     * 自动启动消费者
     */
    @EventListener(ApplicationReadyEvent.class)
    public void autoStart() {
        if (!running) {
            startConsumer();
            System.out.println("Redis消息队列消费者已自动启动...");
        }
    }

    /**
     * 启动消费者
     */
    public synchronized void startConsumer() {
        if (running) {
            System.out.println("消费者已经在运行中");
            return;
        }

        running = true;

        consumerThread = new Thread(() -> {
            System.out.println("消息消费者线程启动，开始监听队列: " + messageQueueName);

            while (running && !Thread.currentThread().isInterrupted()) {
                try {
                    // 阻塞获取消息，超时时间30秒
                    Object messageObj = redisTemplate.opsForList().rightPop(
                            messageQueueName, 30, TimeUnit.SECONDS);

                    // 检查是否应该退出
                    if (!running) {
                        System.out.println("收到停止信号，退出消费循环");
                        break;
                    }

                    if (messageObj instanceof Message) {
                        Message message = (Message) messageObj;
                        processMessage(message);
                    }

                } catch (RedisConnectionFailureException e) {
                    System.err.println("Redis连接失败: " + e.getMessage());
                    if (!running) break;
                    safeSleep(5000); // Redis连接失败时等待重试
                } catch (Exception e) {
                    System.err.println("消费消息时发生未知错误: " + e.getMessage());
                    errorCount.incrementAndGet();
                    if (!running) break;
                    safeSleep(1000);
                }
            }

            running = false;
            System.out.println("消息消费者线程已退出");

        }, "Redis-Message-Consumer");

        consumerThread.setDaemon(false); // 非守护线程，确保应用关闭时能正常结束
        consumerThread.start();
    }

    /**
     * 停止消费者
     */
    public synchronized void stopConsumer() {
        if (!running) {
            System.out.println("消费者未运行");
            return;
        }

        System.out.println("正在停止消息消费者...");
        running = false;

        if (consumerThread != null && consumerThread.isAlive()) {
            consumerThread.interrupt();
            try {
                // 等待线程结束，最多等待10秒
                consumerThread.join(10000);
                if (consumerThread.isAlive()) {
                    System.err.println("消费者线程未能正常结束，可能发生阻塞");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println("等待消费者线程结束时被中断");
            }
        }

        System.out.println("消息消费者已停止");
    }

    /**
     * 安全睡眠，检查中断状态
     */
    private void safeSleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("睡眠被中断", e);
        }
    }

    /**
     * 处理消息
     */
    private void processMessage(Message message) {
        try {
            System.out.println("开始处理消息: " + message.getId());

            // 模拟业务处理
            // 这里替换为您的实际业务逻辑
            Thread.sleep(100);

            processedCount.incrementAndGet();
            lastProcessTime = System.currentTimeMillis();

            System.out.println("消息处理完成: " + message.getId());

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("消息处理被中断: " + message.getId(), e);
        } catch (Exception e) {
            errorCount.incrementAndGet();
            System.err.println("处理消息失败: " + message.getId() + ", 错误: " + e.getMessage());
            throw e; // 重新抛出异常，让上层处理
        }
    }

    /**
     * 获取消费者状态
     */
    public ConsumerStatus getStatus() {
        return new ConsumerStatus(
                running,
                processedCount.get(),
                errorCount.get(),
                lastProcessTime,
                lastProcessTime > 0 ? System.currentTimeMillis() - lastProcessTime : -1
        );
    }

    /**
     * 优雅关闭
     */
    @EventListener(ContextClosedEvent.class)
    public void onApplicationShutdown() {
        System.out.println("应用关闭，正在停止消息消费者...");
        stopConsumer();
    }

    /**
     * 状态对象
     */
    @Data
    public static class ConsumerStatus {
        private final boolean running;
        private final long processedCount;
        private final long errorCount;
        private final long lastProcessTime;
        private final long inactiveDuration;

        // 构造函数、getter方法...
    }
}
